package club.hicode.daydayup.zookeeper.distribute1;

import cn.hutool.core.thread.ThreadUtil;
import cn.hutool.core.util.RandomUtil;
import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

/**
 * 测试 Zookeeper 的分布式锁
 *
 * @author Liu Chunfu
 * @date 2018-06-06 下午5:00
 **/
public class ZooDistributeLock implements Watcher {

    private static final Logger LOG = LoggerFactory.getLogger(ZooDistributeLock.class);

    private static final String LOCK_PATH = "/lock";

    // 模拟开启的线程数
    private static final int THREAD_NUM = 5;

    // 用于等待所有线程都连接成功后再执行任务
    private static CountDownLatch startFlag = new CountDownLatch(1);


    //process 监听的是这个 zk 的变化
    private ZooKeeper zk = null;

    /**
     * 当前路径
     */
    private String currentPath;

    /**
     * 上一个被锁的 lock
     */
    private String preLockPath;


    @Override
    public void process(WatchedEvent event) {
        //
        Event.KeeperState state = event.getState();
        // 事件类型
        Event.EventType type = event.getType();

        if (Event.KeeperState.SyncConnected == state) {
            if (Event.EventType.None == type) {
                // 标识连接成功
                LOG.info("{}-成功连接上ZK服务器", Thread.currentThread().getId());
                startFlag.countDown();
            }

            if (Event.EventType.NodeDeleted == type && event.getPath().equals(this.preLockPath)) {
                LOG.info("节点" + this.preLockPath + "的锁已经被释放");
                try {
                    // 上一个节点释放了,当前节点去获取锁
                    getLock();
                } catch (KeeperException | InterruptedException e) {
                    LOG.error(e.getMessage(), e);
                }
                System.out.println("我被惊醒了！");
            }
        }

    }

    /**
     * 连接到 ZK
     *
     * @throws IOException
     */
    public void connection() throws IOException, InterruptedException {
        zk = new ZooKeeper("39.108.231.92:2181", 5000, this);
        // 等待连接成功后再执行下一步操作
        startFlag.await();
    }

    // 创建节点,并初始化当前路径
    public void createNode() throws KeeperException, InterruptedException, UnsupportedEncodingException {
        this.currentPath = this.zk.create(LOCK_PATH, "".getBytes("UTF-8"), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode
                .EPHEMERAL_SEQUENTIAL);
    }

    /**
     * 核心方法
     *
     * @throws KeeperException
     * @throws InterruptedException
     */
    public void getLock() throws KeeperException, InterruptedException {
        if (minNode()) {
            doSomething();
            releaseLock();
        }
        //任务没有完成就不释放这个链接
        //System.out.println(Thread.currentThread().getId() + " 结束！");
    }

    /**
     * 当前是否为最小节点
     *
     * @return
     */
    private boolean minNode() {
        // 当前序号
        try {
            // 获取上一个节点
            initLockPath();
            // 判断前一个节点存在不存在,如果存在,则表示当前节点不是最小节点
            // zk.getData(this.preLockPath, this, new Stat());
            zk.getData(this.preLockPath, true, new Stat());
            LOG.info(this.currentPath + " 不是最小值,没有获取锁,等待 " + this.preLockPath + " 释放锁");
            return false;
        } catch (KeeperException e) {
            //获取的时候通过抛出异常来判定是否是最小值
            LOG.info(this.currentPath + " 是最小值,获得锁");
            return true;
        } catch (InterruptedException e) {
            LOG.error(e.getMessage(), e);
        }
        return true;
    }

    private void doSomething() {
        LOG.info("处理业务逻辑...");
        ThreadUtil.sleep(RandomUtil.randomInt(5), TimeUnit.SECONDS);
    }

    /**
     * 释放锁并关闭连接
     *
     * @throws KeeperException
     * @throws InterruptedException
     */
    private void releaseLock() throws KeeperException, InterruptedException {
        if (this.zk != null) {
            LOG.info(this.currentPath + " 业务处理完毕,释放锁...");
            zk.delete(this.currentPath, -1);
            this.zk.close();
            LOG.info(Thread.currentThread().getName() + "关闭 zookeeper 连接");
        }
    }

    /**
     * 将节点的大小减少1
     * 初始化 lockpath
     */
    private void initLockPath() {
        //System.out.println(this.currentPath);
        int currentSeq = Integer.parseInt(this.currentPath.substring(LOCK_PATH.length()));
        // 上一个序号
        int preSeq = currentSeq - 1;
        String preSeqStr = String.valueOf(preSeq);
        //todo 临时节点是10位，这个是确定的嘛？
        while (preSeqStr.length() < 10) {
            preSeqStr = "0" + preSeqStr;
        }
        this.preLockPath = LOCK_PATH + preSeqStr;
        // 加入中间缺了一个数字不就出问题了嘛？
    }

    @Override
    public String toString() {
        return "ZooDistributeLock{" +
                "zk=" + zk +
                ", currentPath='" + currentPath + '\'' +
                ", preLockPath='" + preLockPath + '\'' +
                '}';
    }
}
